package cn.doitedu.datayi.etl

import org.apache.spark.sql.SparkSession

/**
 * @author 涛哥
 * @nick_name "deep as the sea"
 * @contact qq:657270652 wx:doit_edu
 * @site www.doitedu.cn
 * @date 2021-08-06
 * @desc 设备账号绑定计算任务
 */
object DeviceAccountBind {

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .appName("设备账号绑定计算任务")
      .master("local")
      .config("spark.sql.shuffle.partitions","1")
      .enableHiveSupport()
      .getOrCreate()


    // 1. 读取当天日志表
    val curDayLog = spark.read
      .table("ods23.app_event_log")
      .where("   dt='2021-08-04'  ")
      .selectExpr("deviceid", "if(account='',null,account) as account", "sessionid", "timestamp")

    // 2. 聚合当天日志，得出 每一种  “设备-账号” 组合的会话数
    curDayLog.createTempView("curday_log")
    val curDayCombineSessionCnt = spark.sql(
      """
        |
        |select
        |  deviceid,
        |  account,
        |  100*count(distinct sessionid) as score,
        |  max(timestamp) as last_login
        |from curday_log
        |group by deviceid,account
        |
        |""".stripMargin)


    // 3. 加载前日的绑定评分表
    curDayCombineSessionCnt.createTempView("cur_agg")
    spark.read.table("dws23.device_account_bind_score").where("dt='2021-08-03'").createTempView("pre")


    // 4. 拿当日聚合结果 join  前日评分表
    /**
     * 前日出现的组合，今日没出现，则分数衰减
     * 前日出现的今日，今日也出现，则分数累加，并更新时间戳
     * 前日没出现的组合，今日出现，则取今日的得分
     */
    spark.sql(
      """
        |
        |insert into table dws23.device_account_bind_score partition (dt='2021-08-04')
        |select
        |
        |  nvl(cur.deviceid,pre.deviceid) as  deviceid,
        |  nvl(cur.account,pre.account)    as  account,
        |  case
        |    when  cur.score is not null and pre.score is not null  then cur.score+pre.score
        |    when  cur.score is not null and pre.score is null  then cur.score
        |    when  cur.score is null and pre.score is not null  then pre.score * 0.6
        |  end as score,
        |  nvl(cur.last_login,pre.last_login)   as   last_login
        |from
        |  cur_agg cur
        |full join
        |  pre
        |on cur.deviceid=pre.deviceid and if(cur.account is null,'',cur.account)=if(pre.account is null,'',pre.account)
        |
        |""".stripMargin)


    spark.close()

  }

}
